package com.atguigu.edu.app.dws;

import bean.UserBehavirFunnelBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import util.DateFormatUtil;
import util.MyClickhouseUtil;
import util.MyKafkaUtil;

import java.time.Duration;


public class DwsUserUserBehavirFunnelWindow {
    public static void main(String[] args) throws Exception {
        //1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //2.检查点相关设置
        //3.从kafka的dwd_traffic_page_log主题中读取数据数据
        String pageLog = "dwd_traffic_page_log";
        String paySuc = "dwd_trade_pay_suc";
        String cartAdd = "dwd_trade_cart_add";
        String orderDetail = "dwd_trade_order_detail";
        String groupId = "user_behavir_funnel_group";
        DataStreamSource<String> pageLogDS = env.addSource(MyKafkaUtil.getKafkaConsumer(pageLog, groupId));
        //4.转换数据类型，将jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> pageObj = pageLogDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String jsonStr) throws Exception {
                return JSON.parseObject(jsonStr);
            }
        });
        //5.过滤出首页和详情页的数据
        SingleOutputStreamOperator<JSONObject> filterDS = pageObj.filter(new FilterFunction<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObj) throws Exception {
                String pageId = jsonObj.getJSONObject("page").getString("page_id");
                return "home".equals(pageId) || "course_detail".equals(pageId);
            }
        });
        //6.设置watermark生成策略指定事件时间字段
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = filterDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                        return jsonObj.getLong("ts");
                    }
                }));
        //6.按照mid分组
        KeyedStream<JSONObject, String> keyedDS = withWatermarkDS.keyBy(data -> data.getJSONObject("common").getString("mid"));
        //7.转换数据类型 jsonObj-实体类（使用状态编程，独立用户）
        SingleOutputStreamOperator<UserBehavirFunnelBean> pageProcessDS = keyedDS.process(new KeyedProcessFunction<String, JSONObject, UserBehavirFunnelBean>() {
            //声明状态，存放首页的上次访问日期
            private ValueState<String> homeLastVisitDate;
            //声明状态，存放详情页的上次访问日期
            private ValueState<String> detailLastVisitDate;

            //在声明周期方法中，给状态赋值，设置状态的ttl（1天）
            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> homeStateDescripter = new ValueStateDescriptor<String>("homeLastVisitDate", String.class);
                homeStateDescripter.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).build());
                ValueStateDescriptor<String> detailStateDescripter = new ValueStateDescriptor<String>("detailLastVisitDate", String.class);
                detailStateDescripter.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).build());
                homeLastVisitDate = getRuntimeContext().getState(homeStateDescripter);
                detailLastVisitDate = getRuntimeContext().getState(detailStateDescripter);
            }

            //每来一条数据，判断是首页还是详情页
            //将状态中的上次访问日期拿到
            //如果是首页，判断首页访问的状态中 如果是空||上次访问日期与今天不是同一天，独立浏览数加一，并更新状态中的日期
            //如果是详情页，判断详情页访问的状态中 如果是空||上次访问日期与今天不是同一天，独立浏览数加一，并更新状态中的日
            @Override
            public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, UserBehavirFunnelBean>.Context ctx, Collector<UserBehavirFunnelBean> out) throws Exception {
                String pageId = jsonObj.getJSONObject("page").getString("page_id");
                String curDate = DateFormatUtil.toDate(jsonObj.getLong("ts"));
                UserBehavirFunnelBean userBehavirFunnelBean = new UserBehavirFunnelBean(
                        "",
                        "",
                        0L,
                        0L,
                        0L,
                        0L,
                        0L,
                        0L
                );
                if ("home".equals(pageId)) {
                    String homeState = homeLastVisitDate.value();
                    if (StringUtils.isEmpty(homeState) || !curDate.equals(homeState)) {
                        userBehavirFunnelBean.setHomeCt(1L);
                        homeLastVisitDate.update(curDate);
                    }
                }
                if ("course_detail".equals(pageId)) {
                    String detailState = detailLastVisitDate.value();
                    if (StringUtils.isEmpty(pageId) || !curDate.equals(detailState)) {
                        userBehavirFunnelBean.setGoodDetailCt(1L);
                        detailLastVisitDate.update(curDate);
                    }
                }
                out.collect(userBehavirFunnelBean);
            }
        });

        //8.从kafka的支付主题读取数据 dwd_trade_pay_suc
        DataStreamSource<String> paySucDS = env.addSource(MyKafkaUtil.getKafkaConsumer(paySuc, groupId));
        //9.转换数据结构 jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> payJsonDS = paySucDS.map(JSON::parseObject);
        //10.指定水位线和时间戳
        SingleOutputStreamOperator<JSONObject> payWithWatermarkDS = payJsonDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                        return jsonObj.getLong("ts") * 1000;
                    }
                }));

        //11.按照uid分组
        KeyedStream<JSONObject, String> keyedPayDS = payWithWatermarkDS.keyBy(data -> data.getJSONObject("common").getString("uid"));

        //12.使用状态编程，独立用户，转换数据类型 jsonObj-实体类
        SingleOutputStreamOperator<UserBehavirFunnelBean> payProcessDS = keyedPayDS.process(
                new KeyedProcessFunction<String, JSONObject, UserBehavirFunnelBean>() {
                    //声明状态，存放上次支付成功的日期
                    private ValueState<String> payLastDateState;

                    @Override
                    //给状态赋值，设置ttl
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("payLastDate", String.class);
                        stateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).build());
                        payLastDateState = getRuntimeContext().getState(stateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, UserBehavirFunnelBean>.Context ctx, Collector<UserBehavirFunnelBean> out) throws Exception {
                        Long ts = jsonObj.getLong("ts");
                        String curDate = DateFormatUtil.toDate(ts*1000);
                        String PayLastDate = payLastDateState.value();
                        if(StringUtils.isEmpty(PayLastDate)||!curDate.equals(payLastDateState)){
                            UserBehavirFunnelBean payBean = new UserBehavirFunnelBean(
                                    "",
                                    "",
                                    0L, 0L, 0L, 0L, 1L,
                                    0L
                            );
                            payLastDateState.update(curDate);
                            out.collect(payBean);
                        }

                    }
                }
        );
        //13.从kafka的加购主题读取数据 dwd_trade_cart_add
        DataStreamSource<String> cartAddDS = env.addSource(MyKafkaUtil.getKafkaConsumer(cartAdd, groupId));
        //14.转换数据类型 jsonstr-jsonObj
        SingleOutputStreamOperator<JSONObject> cartJsonObjDS = cartAddDS.map(JSON::parseObject);
        //15.指定水位线和时间戳
        SingleOutputStreamOperator<JSONObject> cartWithWatermarkDS = cartJsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong("ts")*1000;
                            }
                        })
        );
        //17.按照uid分组
        KeyedStream<JSONObject, String> cartKeyedDS = cartWithWatermarkDS.keyBy(data -> data.getJSONObject("common").getString("uid"));
        //18.使用状态编程，判断独立加购人数，并转为实体类
        SingleOutputStreamOperator<UserBehavirFunnelBean> cartProcessDS = cartKeyedDS.process(new KeyedProcessFunction<String, JSONObject, UserBehavirFunnelBean>() {
            private ValueState<String> cartLastDateState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("cartLastDateState", String.class);
                stateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).build());
                cartLastDateState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, UserBehavirFunnelBean>.Context ctx, Collector<UserBehavirFunnelBean> out) throws Exception {
                Long ts = jsonObj.getLong("ts");
                String curDate = DateFormatUtil.toDate(ts * 1000);
                String lastDate = cartLastDateState.value();
                if (StringUtils.isEmpty(lastDate) || !curDate.equals(lastDate)) {
                    UserBehavirFunnelBean userBehavirFunnelBean = new UserBehavirFunnelBean(
                            "",
                            "",
                            0L, 0L, 1L, 0L, 0L,
                            0L
                    );
                    cartLastDateState.update(curDate);
                    out.collect(userBehavirFunnelBean);
                }
            }
        });

        //19.从kafka的下单主题读取数据 dwd_trade_order_detail
        DataStreamSource<String> orderDetailDS = env.addSource(MyKafkaUtil.getKafkaConsumer(orderDetail, groupId));
        //20.转换数据格式 jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> orderJsonObjDS = orderDetailDS.map(JSON::parseObject);
        //21.指定水位线和时间戳
        SingleOutputStreamOperator<JSONObject> orderWithWatermarkDS = orderJsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObj , long recordTimestamp) {
                        return jsonObj.getLong("ts")*1000;
                    }
                })
              );
        //22.按照uid分组
        KeyedStream<JSONObject, String> orderKeyedDS = orderWithWatermarkDS.keyBy(data -> data.getJSONObject("common").getString("uid"));
        //23.使用状态编程，判断独立下单人数，并转为实体类对象
        SingleOutputStreamOperator<UserBehavirFunnelBean> orderProcessDS = orderKeyedDS.process(new KeyedProcessFunction<String, JSONObject, UserBehavirFunnelBean>() {
            //声明状态
            private ValueState<String> orderLastDateState;

            //给状态赋值，设置生命周期

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("orderLastDateState", String.class);
                valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                orderLastDateState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            //判断独立下单用户
            public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, UserBehavirFunnelBean>.Context ctx, Collector<UserBehavirFunnelBean> out) throws Exception {
                Long ts = jsonObj.getLong("ts");
                String curDate = DateFormatUtil.toDate(ts * 1000);
                String orderLastDate = orderLastDateState.value();
                if (StringUtils.isEmpty(orderLastDate) || !curDate.equals(orderLastDate)) {
                    UserBehavirFunnelBean userBehavirFunnelBean = new UserBehavirFunnelBean(
                            "",
                            "",
                            0L, 0L, 0L, 1L, 0L,
                            0L
                    );
                    orderLastDateState.update(curDate);
                    out.collect(userBehavirFunnelBean);
                }

            }
        });

        //24.将四条流union
        DataStream<UserBehavirFunnelBean> unionDS = pageProcessDS.union(payProcessDS, payProcessDS, cartProcessDS);
        //25.开窗（全窗口函数）
        AllWindowedStream<UserBehavirFunnelBean, TimeWindow> windowDS = unionDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //26.聚合
        SingleOutputStreamOperator<UserBehavirFunnelBean> reduceDS = windowDS.reduce(
                new ReduceFunction<UserBehavirFunnelBean>() {
                    @Override
                    public UserBehavirFunnelBean reduce(UserBehavirFunnelBean value1, UserBehavirFunnelBean value2) throws Exception {
                        value1.setHomeCt(value1.getHomeCt() + value2.getHomeCt());
                        value1.setGoodDetailCt(value1.getGoodDetailCt() + value2.getGoodDetailCt());
                        value1.setCartUuCt(value1.getCartUuCt() + value2.getCartUuCt());
                        value1.setOrderUuCt(value1.getOrderUuCt() + value2.getCartUuCt());
                        value1.setPaySucCt(value1.getPaySucCt() + value2.getPaySucCt());
                        return value1;
                    }
                },
                new AllWindowFunction<UserBehavirFunnelBean, UserBehavirFunnelBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<UserBehavirFunnelBean> input, Collector<UserBehavirFunnelBean> out) throws Exception {
                        for (UserBehavirFunnelBean userBehavirFunnelBean : input) {
                            userBehavirFunnelBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                            userBehavirFunnelBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                            userBehavirFunnelBean.setTs(System.currentTimeMillis());
                        }
                    }
                }
        );
        //27.将聚合的结果写到clickhouse
        reduceDS.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_user_behavir_funnel_window values(?,?,?,?,?,?,?,?)"));
        env.execute();
    }
}
